package com.tool.engine.export.capacity;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;


import javax.annotation.PostConstruct;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Data
public class ExportTaskCapacityLimit  implements CapacityLimit{

    private final ReentrantLock lock = new ReentrantLock();

    private final Condition fullCondition = lock.newCondition();

    private final AtomicLong capacity = new AtomicLong(0);

    private long threshold;

    private volatile boolean exitLock = false;

    private int factor;

    public void addCapacity(long size) {
        capacity.updateAndGet(pre -> pre + size);
    }

    public void removeCapacity(long size) {
        long total = capacity.updateAndGet(pre -> pre - size);
        if (total <= threshold && exitLock) {
            try {
                lock.lock();
                if(lock.hasWaiters(fullCondition)) {
                    fullCondition.signalAll();
                    exitLock = false;
                    log.info("threadName:{} exitLock:{} capacity:{} start awakening paused threads", Thread.currentThread().getName(), exitLock, capacity.get());
                }
            } finally {
                lock.unlock();
            }
        }
    }

    public void tryLock(long size) {
        if ((capacity.get()+ size) >= threshold) {
            try {
                lock.lock();
                log.info("threadName:{}, capacity:{} limit reached", Thread.currentThread().getName(), capacity.get());
                while (capacity.get() >= threshold) {
                    exitLock = true;
                    log.info("threadName:{} capacity:{} starts to sleep exitLock:{}", Thread.currentThread().getName(), capacity.get(), exitLock);
                    fullCondition.awaitUninterruptibly();
                    log.info("threadName:{}, capacity:{} awakened exitLock:{}", Thread.currentThread().getName(), capacity.get(), exitLock);
                }
            } finally {
                lock.unlock();
            }
        }
    }

    @PostConstruct
    public void afterPropertiesSet() throws Exception {
        long maxMemory = Runtime.getRuntime().maxMemory();
        log.info("JVM memory:{} M", maxMemory / 1024 / 1024);
        threshold = maxMemory * factor /100;
        log.info("export task capacity limit:{} M", threshold / 1024 / 1024);
    }
}
